package com.atguigu.day07;

import com.atguigu.bean.Event;
import com.atguigu.day03.Flink01_Source_Customer;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink11_KeyedState_MapState {

    //我们要计算的是每一个url在每一个窗口中的pv数据
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Event> streamOperator = env
                .addSource(new Flink01_Source_Customer.ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {

                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        }));

        //将相同url的数据聚合到一块
        streamOperator.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event value) throws Exception {
                return value.url;
            }
        })
                .process(new MyFakeWindow(5000))

                .print()
        ;

        env.execute();

    }

    public static class MyFakeWindow extends KeyedProcessFunction<String,Event,String>{

        //声明一个MapState (k:=>窗口的开始时间) （v：=>当前窗口结算pv的结果）
        private MapState<Long, Integer> mapState;


        private Integer windowSize;

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Integer>("map-State", Long.class, Integer.class));
        }

        public MyFakeWindow(Integer windowSize){
            this.windowSize = windowSize;
        }

        @Override
        public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
            //1.计算属于这个数据的窗口开始时间
            Long windowStart = value.timestamp - (value.timestamp - 0 + windowSize) % windowSize;
            //2.计算窗口的结束时间
            Long windowEnd = windowStart + windowSize;

            //3.判断状态中是否有当前窗口计算的结果
            if (mapState.contains(windowStart)){
                //存在之前计算的结果
                Integer lastPvCount = mapState.get(windowStart);
                lastPvCount += 1;
                //再写入状态
                mapState.put(windowStart, lastPvCount);
            }else {
                //之前没有计算结果
                mapState.put(windowStart, 1);
            }

            //定义一个定时器，用来控制窗口的关闭以及触发计算
            ctx.timerService().registerEventTimeTimer(windowEnd-1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //窗口开始时间
            Long windowStart = timestamp + 1 - windowSize;
            Long windowEnd = windowStart + windowSize;
            out.collect("窗口：["+windowStart+","+windowEnd+") ====>pv:"+mapState.get(windowStart));

            //触发器触发意味着出了触发窗口计算之外，还要关窗
            mapState.remove(windowStart);
        }
    }
}
